package org.nbict.iot.trash.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.tuple.Fields;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.task.realtime.filter.AuthFilter;
import org.nbict.iot.task.realtime.filter.UploadFilter;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.task.realtime.func.RealtimeResolver;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;
import org.nbict.iot.trident.mogo.MultiQueryFilterCreator;
import org.nbict.iot.trident.mogo.mapper.MongoMapper;
import org.nbict.iot.trident.mogo.mapper.impl.SimpleMongoMapper;
import org.nbict.iot.trident.mogo.mapper.impl.SimpleMongoUpdateMapper;
import org.nbict.iot.trident.mogo.state.MongoState;
import org.nbict.iot.trident.mogo.state.MongoStateFactory;
import org.nbict.iot.trident.mogo.state.MongoStateUpdater;

/**
 * Created by songseven on 18/6/25.
 */
@Deprecated
public class RealtimeTrident {


    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        String url = PropertyUtil.getProperty("mongo.url");
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String admin = PropertyUtil.getProperty("mongo.admin.name");
        String collectionName = "car";
        String collectionName1 = "car_authc";

        TridentTopology topology = new TridentTopology();

        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server,
                        // topic transit-.*-simple
                "iot-car-group-realtime-update"); //group id
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-parsed", kafkaSpout);

        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");

        Stream stage = stream.each(new Fields("platform_id", "packet"), new PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header", "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action", "timestamp", "upload")); //去除packet元组

        //实时上报信息流
        Stream uploadStream = stage.each(outputFields, new UploadFilter()).partitionBy(new Fields("car_vin"))
                .toStream(); //2 or 3

        //1 or 4 or 5 or 6
        //更新车辆实时信息
        MongoMapper mapper = new SimpleMongoUpdateMapper().withFields
                ("platform_id", "car_vin", "timestamp", "01", "02", "03",
                        "04", "05", "06", "07", "08", "09") //更新的字段
                .withFilter(new MultiQueryFilterCreator().eq("car_vin"));//.lt("timestamp")); //过滤条件

        MongoState.Options options = new MongoState.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName)
                .withMapper(mapper);

        //持久化字段
        Fields persistFields = new Fields("platform_id", "car_vin", "timestamp", "01", "02", "" +
                "03", "04", "05", "06", "07", "08", "09");
        //更新实时数据
        uploadStream.each(new Fields("platform_id", "car_vin", "timestamp",
                "upload"), new RealtimeResolver(), new Fields("01", "02",
                "03", "04", "05", "06", "07", "08", "09"))
                //持久化
                .partitionPersist(new MongoStateFactory(options), persistFields, //持久化
                        new MongoStateUpdater(), new Fields());

        //登入登出流
        Stream authStream = stage.each(outputFields, new AuthFilter());

        MongoMapper authcMapper = new SimpleMongoMapper().withFields
                ("platform_id", "car_vin", "timestamp", "action"); //更新的字段
        MongoState.Options authcOptions = new MongoState.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName1)
                .withMapper(authcMapper);
        authStream.partitionPersist(new MongoStateFactory(authcOptions),
                new Fields("platform_id", "car_vin", "timestamp", "action"),
                new MongoStateUpdater(), new Fields() ); //记录登录登出

        Config config = new Config();
        config.setMaxSpoutPending(3);

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-realtime", config, topology.build());
        }
    }
}
